package org.devgframwork.datasync.core.support.init.rabbitmq;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import lombok.extern.slf4j.Slf4j;
import org.devgframwork.datasync.core.entity.RabbitConfig;
import org.devgframwork.datasync.core.entity.TriggerConfig;
import org.devgframwork.datasync.core.support.DataSourceMemory;
import org.devgframwork.datasync.core.support.init.AbstractQueueInit;
import org.devgframwork.datasync.core.utils.CommonUtil;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static org.devgframwork.datasync.core.support.DataSourceMemory.triggerConfigMap;

/**
 * @Description
 * @auther dh
 * @date 2021/12/26 17:25
 */
@Slf4j
public class RabbitQueueInit extends AbstractQueueInit<RabbitConfig, TriggerConfig> {

    public static final String suffixExchange = "_exchange";

    @Override
    public void invoke(RabbitConfig queueConfig, List<TriggerConfig> triggerConfigList) {
        // 第一步初始化 queue配置
        if (null == queueConfig || CommonUtil.isNullOrEmpty(queueConfig.getConnectionName()) ||
                CommonUtil.isNullOrEmpty(queueConfig.getHost()) || CommonUtil.isNullOrEmpty(queueConfig.getPassword())) {
            log.warn("rabbitQueueConfig error");
            return;
        }
        if (null == triggerConfigList || triggerConfigList.size() <= 0) {
            log.warn("rabbitQueueList is Empty");
            return;
        }
        Channel channel = getChannel(queueConfig);
        if (null == channel) {
            log.error("rabbitQueue channel cannot be null");
            return;
        }
        DataSourceMemory.connectionChannelMap.putIfAbsent(queueConfig.getConnectionName(), channel);
        String exchange = queueConfig.getConnectionName() + suffixExchange;
        for (TriggerConfig triggerConfig : triggerConfigList) {
            try {
                initRabbitMq(channel, triggerConfig, exchange);
                RabbitQueueConsumerInit.getInstance(triggerConfig).initConsumer();
                String connectionName = triggerConfig.getConnectionName();
                String tableName = triggerConfig.getTableName();
                String mapKey = connectionName.concat(":").concat(tableName);
                triggerConfigMap.put(mapKey, triggerConfig);
                log.info("init rabbitmq {} success", triggerConfig.getTableName());
            } catch (Exception ex) {
                log.error("init rabbitmq error,{}", ex.getMessage());
            }

        }

    }

    private Channel getChannel(RabbitConfig config) {
        Channel channel = null;
        try {
            ConnectionFactory connectionFactory = getConnectionFactory(config);
            int nThreads = (config.getThreadPoolSize() == null ? 5 : config.getThreadPoolSize());
            ExecutorService es = Executors.newFixedThreadPool(nThreads);
            Connection connection = connectionFactory.newConnection(es);
            if (connection == null) {
                return null;
            }
            channel = connection.createChannel();
        } catch (Exception ex) {
            ex.printStackTrace();
            log.error("" + ex.getMessage());
        }
        return channel;
    }

    private ConnectionFactory getConnectionFactory(RabbitConfig config) {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(config.getHost());
        connectionFactory.setPort(config.getPort());
        connectionFactory.setUsername(config.getUsername());
        connectionFactory.setPassword(config.getPassword());
        if (!CommonUtil.isNullOrEmpty(config.getVirtualHost())) {
            connectionFactory.setVirtualHost(config.getVirtualHost());
        }
        Integer multiple = 5;
        connectionFactory.setAutomaticRecoveryEnabled(true);
        connectionFactory.setNetworkRecoveryInterval(60000 * multiple);
        connectionFactory.setRequestedHeartbeat(2 * multiple);
        connectionFactory.setConnectionTimeout(60000 * multiple);
        connectionFactory.setHandshakeTimeout(60000 * multiple);
        return connectionFactory;
    }


    public void initRabbitMq(Channel channel, TriggerConfig triggerConfig, String exchange) throws Exception {
        String queueName = "ds_".concat(triggerConfig.getTableName());
        channel.exchangeDeclare(exchange, BuiltinExchangeType.DIRECT, true, false, null);  // 声明交换器
        String deadQueueName = queueName.concat("_retry");
        String errorQueueName = queueName.concat("_error");
        triggerConfig.setExecuteName(triggerConfig.getConnectionName().concat(triggerConfig.getTableName()));
        triggerConfig.setExchange(exchange);
        triggerConfig.setNormalQueue(queueName);
        triggerConfig.setRetryQueue(deadQueueName);
        triggerConfig.setErrorQueue(errorQueueName);
        initRabbitMq(exchange, queueName, deadQueueName, errorQueueName, channel);
    }

    public void initRabbitMq(String changeName, String queueName, String deadQueueName, String errorQueueName, Channel channel) throws Exception {
        // 声明2个交换器,正常交换器和重试队列交换器
        // channel.exchangeDeclare(changeName, BuiltinExchangeType.DIRECT, true, false, null);
        // 声明2个队列,正常队列和重试队列
        Map<String, Object> args = new HashMap<>();
        args.put("x-dead-letter-exchange", changeName);
        args.put("x-dead-letter-routing-key", queueName);
        channel.queueDeclare(deadQueueName, true, false, false, args);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueDeclare(errorQueueName, true, false, false, null);
        //将正常队列和正常交换器绑定到一起,重试队列和重试队列交换器绑定到一起
        //正常队列绑定到错误交换器里面,如果消息到达重试队列队列,
        //路由键的名称和队列名称一样
        channel.queueBind(queueName, changeName, queueName);
        channel.queueBind(deadQueueName, changeName, deadQueueName);
        channel.queueBind(errorQueueName, changeName, errorQueueName);
    }

}
